/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by WangYunlai on 2022/07/01.
//

#include "common/log/log.h"
#include "sql/operator/project_physical_operator.h"
#include "storage/record/record.h"
#include "storage/table/table.h"

RC ProjectPhysicalOperator::open(Trx *trx)
{
  if (children_.empty()) {
    return RC::SUCCESS;
  }

  PhysicalOperator *child = children_[0].get();
  RC rc = child->open(trx);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open child operator: %s", strrc(rc));
    return rc;
  }

  return RC::SUCCESS;
}

RC ProjectPhysicalOperator::next()
{
  if (children_.empty()) {
    return RC::RECORD_EOF;
  }
  return children_[0]->next();
}

RC ProjectPhysicalOperator::close()
{
  if (!children_.empty()) {
    children_[0]->close();
  }
  return RC::SUCCESS;
}
Tuple *ProjectPhysicalOperator::current_tuple()
{
  if (is_none_func_ == nullptr) {
    is_none_func_ = new bool(true);
    const std::vector<AggFuncType>& types = tuple_.func_types();
    for(AggFuncType type : types) {
      if (type != FUNC_NONE) {
        *is_none_func_ = false;
        break;
      }
    }
  }
  // TODO 可能做了一些没有必要的计算，可以考虑进行修改
  if (!(*is_none_func_)) {
    RC rc = RC::SUCCESS;
    const std::vector<TupleCellSpec *> &speces = tuple_.speces();
    std::vector<Value> values;
    Tuple *tuple = children_[0]->current_tuple();
    // 用来存储给avg使用的被除数
    std::unordered_map<Value *, int> value_size;
    const std::vector<AggFuncType> &func_types = tuple_.func_types();
    // 先把当前tuple中的数据取出
    for (int i = 0; i < speces.size(); i++) {
      Value tmp;
      tuple->find_cell(*speces[i], tmp);
      values.push_back(tmp);
      // 把为count聚合函数的初始值赋值
      if (func_types[i] == FUNC_COUNT) {
        values[i].set_int(values[i].attr_type() == NULLS? 0: 1);
      }else if (func_types[i] == FUNC_AVG) {
        value_size.insert(std::pair<Value *, int>(&values[i], values[i].attr_type() == NULLS? 0: 1)); 
      }
    }
    while((rc = next()) == RC::SUCCESS) {
      Tuple *tuple = children_[0]->current_tuple();
      for(int i = 0; i < speces.size(); i++) {
        Value tmp;
        tuple->find_cell(*speces[i], tmp);
        // 为NULL值的数据不参与计算
        if (tmp.attr_type() != NULLS) {
          switch (tuple_.func_types()[i]) {
          case FUNC_MAX: {
            if (values[i].compare(tmp) == RC::LEFT_LT_ANOTHER) {
              values[i].set_value(tmp);
            }
          };break;
          case FUNC_MIN: {
            if (values[i].compare(tmp) == RC::LEFT_GT_ANOTHER) {
              values[i].set_value(tmp);
            }
          }break;
          case FUNC_AVG: {
            value_size[&values[i]];
            if (tmp.attr_type() == FLOATS) {
              values[i].set_float(values[i].get_float() + tmp.get_float());
            }else {
              values[i].set_int(values[i].get_int() + tmp.get_int());
            }
          }break;
          case FUNC_COUNT: {
            values[i].set_int(values[i].get_int() + 1);
          }break;
          default:{
            LOG_WARN("Unacceptable aggregate func");
          }break;
          }
        }
        
      }
    }
    // 计算count和avg
    for (int i = 0; i < speces.size(); i++) {
      if (func_types[i] == FUNC_AVG) {

        Value &val = values[i];
        AttrType type = val.attr_type();
        int size = value_size[&val];
        if (size > 0 && (type == AttrType::INTS || type == AttrType::FLOATS)) {
          values[i].set_float(values[i].get_float() / size);
        }else {
          values[i].set_type(NULLS);
        }
      }
    }
    // 把聚合好的数据装入value_list中
    ValueListTuple *value_list = new ValueListTuple;
    value_list->set_cells(values);
    tuple_.set_tuple(value_list);
  } else {
    tuple_.set_tuple(children_[0]->current_tuple());
  }
  return &tuple_;
}

void ProjectPhysicalOperator::add_projection(const Field *field)
{
  // 对单表来说，展示的(alias) 字段总是字段名称，
  // 对多表查询来说，展示的alias 需要带表名字
  TupleCellSpec *spec = new TupleCellSpec(field->table_name(), field->meta()->name(), field->meta()->name());
  tuple_.add_cell_spec(spec, field->type());
}
